#ifndef SDFS_SERVER_H
#define SDFS_SERVER_H

#include <muduo/base/Atomic.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Thread.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/InetAddress.h>

#include <boost/bind.hpp>

#include <stdlib.h>

#include <sdfs/protobuf/Message.h>
#include <sdfs/config/TrackerConfig.h>
#include <sdfs/protobuf/Codec.h>
#include <sdfs/protobuf/ProtobufDispatcher.h>
#include <sdfs/redis/SyncRedisConnector.h>
#include <sdfs/tracker/StorageServerProxy.h>
#include <sdfs/tracker/FileInfoCodec.h>

using namespace muduo::net;

using namespace sdfs::protobuf;

using namespace sdfs::redis;

namespace sdfs
{

namespace tracker
{

class TrackerServer
{
public:
	TrackerServer(EventLoop* loop, const InetAddress& addr, int threadNum)
		:server_(loop, addr, sdfs::TrackerConfig::name()),
		 dispatcher_(
		 	boost::bind(&TrackerServer::discardMessage, this, _1, _2, _3)
		 ),
		 codec_(
		 	boost::bind(&ProtobufDispatcher::onProtobufMessage, &dispatcher_, _1, _2, _3)
		 ),
		 redisConn_(sdfs::TrackerConfig::redisIp().c_str(), sdfs::TrackerConfig::redisPort()),
		 startTime_(Timestamp::now()),
		 threadNum_(threadNum)
	{
		server_.setConnectionCallback(
			boost::bind(&TrackerServer::onConnection, this, _1));
		server_.setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &codec_, _1, _2, _3));

		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::GroupSeeker>(
			boost::bind(&TrackerServer::onGroupSeek, this, _1, _2, _3));
		dispatcher_.registerProtobufMessageCallback<sdfs::protobuf::FileFindMessage>(
			boost::bind(&TrackerServer::onFindFile, this, _1, _2, _3));
		
		
	}

	void onFindFile(const muduo::net::TcpConnectionPtr& conn,
		const shared_ptr<FileFindMessage> message, muduo::Timestamp timestamp)
	{
		char cmd[64];
		snprintf(cmd, 64, "LRANGE %s 0 -1", message->hashcode().c_str());
		RedisReplyPtr reply = redisConn_.command(cmd);
		ServerStatusPtr status;
		FileFindResultMessage result;
		result.set_id(message->id()+1);
		result.set_hashcode(message->hashcode());
		for (int i = 0; i < reply->size(); ++i)
		{
			muduo::string value = reply->getStrAt(i);
			FileInfoCodec codec(value);
			ServerStatusPtr status2 = StorageServerProxy::getInstance()->findStatus(codec.getStorageName());
			if(status2 == NULL)
			{
				continue;
			}
			if(status == NULL || status2 < status)
			{
				status = status2;

				result.set_result(sdfs::protobuf::FileFindResultMessage::FOUND);
				sdfs::protobuf::Worker* worker = result.mutable_worker();
				worker->set_ip(status->getIp().c_str());
				worker->set_port(status->getPort());

				sdfs::protobuf::Filepiece* piece = result.mutable_piece();
				piece->set_blockid(codec.getBlockId());
				piece->set_offset(codec.getOffset());
				piece->set_length(codec.getLength());

				LOG_DEBUG << "find result: "<< status->getIp() << ":" <<status->getPort() << ","
					<<codec.getBlockId()<<","<<codec.getOffset()<<","<<codec.getLength();
			}
		}
		if (status == NULL)
		{
			result.set_result(sdfs::protobuf::FileFindResultMessage::NOTEXIST);
		}
		codec_.Send(conn, result);
		conn->shutdown();
	}

	void onGroupSeek(const muduo::net::TcpConnectionPtr& conn,
		const GroupSeekerPtr message, muduo::Timestamp timestamp)
	{	
		LOG_DEBUG << "[" << conn->peerAddress().toIpPort() << "] seeks for service";

		sdfs::protobuf::GroupPackage responseMessage;
		responseMessage.set_id(curId_.addAndGet(1));
		responseMessage.set_reqid(message->id());

		if(message->type() == sdfs::protobuf::GroupSeeker::UPLOAD)
		{
			ServerStatusPtr server = StorageServerProxy::getInstance()->findWriteStorage();
			if(server == NULL)
			{
				responseMessage.set_result(sdfs::protobuf::GroupPackage::REFUSE);
				responseMessage.set_reason("No Storage Service found");
			}
			else
			{
				responseMessage.set_result(sdfs::protobuf::GroupPackage::ACCEPT);
				sdfs::protobuf::GroupPackage_Worker* si = responseMessage.add_worker();
				si->set_ip(server->getIp().c_str());
				si->set_port(server->getPort());
			}
		}
		else
		{
			shared_ptr< std::vector<ServerStatusPtr> > storages = 
				StorageServerProxy::getInstance()->packageStorageGroup();

			
			if(storages->size() < 1)
			{
				responseMessage.set_result(sdfs::protobuf::GroupPackage::REFUSE);
				responseMessage.set_reason("No Storage Service found");
			}
			else
			{
				responseMessage.set_result(sdfs::protobuf::GroupPackage::ACCEPT);
				// char ip[32];
				for (uint i = 0; i < storages->size(); ++i)
				{
					/* code */
					//fix me: if a storage had lost during dispatching to client, 
					//         removeStorage() will destroy ServerStatus, getIp() method
					//         may cause core dump.
					sdfs::protobuf::GroupPackage_Worker* si = responseMessage.add_worker();
					si->set_ip(storages->at(i)->getIp().c_str());
					si->set_port(storages->at(i)->getPort());
				}
			}
		}
		
		codec_.Send(conn, responseMessage);
		conn->shutdown();
	}

	

	void discardMessage(const TcpConnectionPtr& conn, 
		const MessagePtr message, Timestamp timestamp)
	{
		LOG_TRACE << "message["<<message->GetTypeName()
			<< "] from: " << conn->peerAddress().toIpPort()
			<< " discard";
	}

	void onConnection(const TcpConnectionPtr& conn)
	{
		LOG_DEBUG << conn->peerAddress().toIpPort() <<
			" -> " << conn->localAddress().toIpPort() <<
			" is " << (conn->connected() ? "UP" : "DOWN");
	}

	void start()
	{
		LOG_DEBUG << "starting tracker server";
		server_.start();
		redisConn_.connect();
	}

	/* data */
private:
	TcpServer server_;
	ProtobufDispatcher dispatcher_;
	ProtobufCodec codec_;
	SyncRedisConnector redisConn_;
	Timestamp startTime_;
	int threadNum_;
	AtomicInt32 curId_;

	
};
}
}

#endif
